package com.flink.es;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;

/**
 * @Author hujf
 * @Date 2022/8/29 10:54
 * @Description TODO
 */
public class HttpDynamicTableSink implements DynamicTableSink {
    private String type;
    private String url;
    private String header;
    private String body;

    private EncodingFormat<SerializationSchema<RowData>> encodingFormat;
    private DataType producedDataType;
    private TableSchema tableSchema;

    public HttpDynamicTableSink(String url, String type, String header, String body, EncodingFormat<SerializationSchema<RowData>> encodingFormat, DataType producedDataType, TableSchema tableSchema) {
        this.encodingFormat = encodingFormat;
        this.producedDataType = producedDataType;
        this.tableSchema = tableSchema;
    }

    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        return requestedMode;
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        final SerializationSchema<RowData> deserializer = encodingFormat.createRuntimeEncoder(
                context,
                producedDataType);
        HttpSinkFunction httpSinkFunction =
                HttpSinkFunction.builder().setUrl(this.url)
                        .setBody(this.body).setDeserializer(deserializer)
                        .setType(this.type).setFields(tableSchema.getFieldNames())
                        .setDataTypes(tableSchema.getFieldDataTypes()).build();
        return SinkFunctionProvider.of(httpSinkFunction);

    }

    @Override
    public DynamicTableSink copy() {
        return new HttpDynamicTableSink(this.url,this.type,this.header,this.body,this.encodingFormat,this.producedDataType,this.tableSchema);
    }

    @Override
    public String asSummaryString() {
        return "HTTP Table Sink";
    }
}
